package com.sunlands.qdcy.tongzhuo.util;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.stereotype.Component;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisCluster;
import redis.clients.jedis.JedisCommands;

import java.util.ArrayList;
import java.util.List;
import java.util.UUID;

/**
 * Redis 可重入分布式锁
 */
@Component
public class RedisDistributedLockImpl implements IRedisDistributedLock {

    private final Logger logger = LoggerFactory.getLogger(RedisDistributedLockImpl.class);

    @Autowired
    private StringRedisTemplate redisTemplate;

    /**
     * 锁前缀
     */
    private static final String ROOT_KEY = "LOCK_";

    /**
     * 过期时间，ms
     */
    private static final long EXPIRE = 5000L;

    /**
     * 最长等待时间，ms
     */
    private static final long WAIT_MILLIS = 10000L;

    /**
     * 重试等待时间，ms
     */
    private static final long SLEEP_MILLIS = 500L;

    /**
     * 最多重试次数
     */
    private static final int RETRIES = 20;

    /**
     * 使用 ThreadLocal 存储 key 的 value 值，防止同步问题
     */
    private ThreadLocal<String> threadLocal = new ThreadLocal<>();

    /**
     * 原子操作释放锁 Lua 脚本
     */
    private static final String LUA_UNLOCK_SCRIPT = "if redis.call(\"get\", KEYS[1]) == ARGV[1] " +
            "then " +
            "return redis.call(\"del\", KEYS[1]) " +
            "else " +
            "return 0 " +
            "end";

    @Override
    public boolean lock(String key) {
        return setLock(key, EXPIRE, WAIT_MILLIS, SLEEP_MILLIS, RETRIES);
    }

    @Override
    public boolean lock(String key, long waitMillis) {
        return setLock(key, EXPIRE, waitMillis, SLEEP_MILLIS, RETRIES);
    }

    @Override
    public boolean lock(String key, long waitMillis, long sleepMillis) {
        return setLock(key, EXPIRE, waitMillis, SLEEP_MILLIS, RETRIES);
    }

    @Override
    public boolean lock(String key, long expire, long waitMillis, long sleepMillis) {
        return setLock(key, expire, waitMillis, sleepMillis, RETRIES);
    }

    @Override
    public boolean lock(String key, long expire, long waitMillis, long sleepMillis, int retries) {
        return setLock(key, expire, waitMillis, sleepMillis, retries);
    }

    /**
     * 获取 Redis 锁
     *
     * @param key         锁名称
     * @param expire      锁过期时间
     * @param retries     最多重试次数
     * @param sleepMillis 重试等待时间
     * @param waitMillis  最长等待时间
     * @return
     */
    private boolean setLock(String key, long expire, long waitMillis, long sleepMillis, int retries) {
        //检查 key 是否为空
        if (key == null || "".equals(key)) {
            return false;
        }

        try {
            long startTime = System.currentTimeMillis();
            key = ROOT_KEY + key;

            //可重入锁判断
            String v = threadLocal.get();
            if (v != null && isReentrantLock(key, v)) {
                return true;
            }

            //获取锁
            String value = UUID.randomUUID().toString();
            while (!this.setNX(key, value, expire)) {
                //超过最大重试次数后获取锁失败
                if (retries-- < 1) {
                    logger.debug("获取锁重试次数过多 key={} retries={}",key, retries);
                    return false;
                }

                //等待下一次尝试
                Thread.sleep(sleepMillis);

                //超过最长等待时间后获取锁失败
                if (System.currentTimeMillis() - startTime > waitMillis) {
                    logger.info("超过最长等待时间后获取锁失败 key={}",key);
                    return false;
                }
            }

            threadLocal.set(value);
            return true;

        } catch (Exception e) {
            logger.error("redis lock get: e={}", e);
            return false;
        }
    }

    /**
     * SET if Not eXists
     */
    private boolean setNX(String key, String value, long expire) {
        String result = (String) redisTemplate.execute((RedisCallback<String>) connection -> {
            JedisCommands commands = (JedisCommands) connection.getNativeConnection();
            return commands.set(key, value, "NX", "PX", expire);
        });

        return "OK".equals(result);
    }

    /**
     * 可重入锁判断
     */
    private boolean isReentrantLock(String key, String v) {
        ValueOperations<String, String> kvValueOperations = redisTemplate.opsForValue();
        String value = kvValueOperations.get(key);
        if (value == null) {
            return false;
        }

        return v.equals(value);
    }

    /**
     * 释放锁
     */
    @Override
    public boolean release(String key) {
        if (key == null || "".equals(key)) {
            return false;
        }

        List<String> keys = new ArrayList<>(1);
        keys.add(ROOT_KEY + key);
        List<String> args = new ArrayList<>(1);
        args.add(threadLocal.get());
        threadLocal.remove();

        try {
            return deleteKey(keys, args);
        } catch (Exception e) {
            logger.info("redis lock release: e={}", e);
        }
        return false;
    }

    /**
     * 删除 redis key
     * <p>集群模式和单机模式执行脚本方法一样，但没有共同的接口
     * <p>使用lua脚本删除redis中匹配value的key，可以避免由于方法执行时间过长而 redis 锁自动过期失效的时候误删其他线程的锁
     */
    private boolean deleteKey(List<String> keys, List<String> args) {
        Object result = redisTemplate.execute((RedisCallback<Object>) connection -> {
            Object nativeConnection = connection.getNativeConnection();
            logger.info("keys={}, args={}", keys, args);
            // 单机模式
            if (nativeConnection instanceof Jedis) {
                return ((Jedis) nativeConnection).eval(LUA_UNLOCK_SCRIPT, keys, args);
            }

            // 集群模式
            else if (nativeConnection instanceof JedisCluster) {
                return ((JedisCluster) nativeConnection).eval(LUA_UNLOCK_SCRIPT, keys, args);
            }

            return 0L;
        });

        return result != null && Long.parseLong(result.toString()) > 0;
    }
}